package com.brid.KafkaProducer;

import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

/**
 * @Author: 司军强
 * @Date: 13:30 2019/12/19
 * @Description: kafka生产者详解
 */
@Slf4j
public class KafkaProducerAnalysis {
    //kfka对外提供的集群地址
    private static final String BROKERLIST = "192.168.223.128:9092";
    //topic 的name
    private static final String TOPIC = "hema";

    /**
     * 初始化kafka的配置信息
     *
     * @return
     */
    public static Properties initConfig() {
        Properties prop = new Properties();
        //设置key序列化
        prop.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        //设置value序列化
        prop.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        //设置kafka的连接地址
        prop.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BROKERLIST);
        //设置kafkaproducer对应的客户端ID ,默认是空
        prop.put("client.id", "producer.client.id.demo");
        return prop;
    }

    /**
     * 初始化新配置 添加拦截器
     *
     * @return
     */
    public static Properties initNewConfig() {
        Properties prop = new Properties();
        prop.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        //设置value序列化
        prop.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        //设置kafka的连接地址
        prop.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BROKERLIST);
        //设置kafka客户端对应的id
        prop.put(ProducerConfig.CLIENT_ID_CONFIG, "producer.client.id.demo");
        //  todo ？？？自定义分区器的使用
        //props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG,DefinePartitioner.class.getName());
        //自定义拦截器
        prop.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, ProducerInterceptorPrefix.class.getName());
        //设置ack回执机制 0 传输最快 安全低 1默认 -1安全 效率低
        prop.put(ProducerConfig.ACKS_CONFIG, 0);
        return prop;
    }

    /**
     * 自定义基础的kafka连接配置
     * 测试使用git
     * @return
     */
    public static Properties initPerferConfig() {
        Properties prop = new Properties();
        prop.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        //设置value序列化
        prop.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        //设置kafka的连接地址
        prop.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BROKERLIST);
        return prop;
    }

    /**
     * 调用方法 往kafka里面发送消息
     *
     * @param args
     */
    public static void main(String[] args) {
        //调用方法
        Properties properties = initNewConfig();
        //创建kafaka的对象 准备发送消息
        KafkaProducer<String, String> producer = new KafkaProducer<>(properties);
        //生成 producerRecord对象 并指定 topic key value
        ProducerRecord record = new ProducerRecord<>(TOPIC, "kafka-demo-001", "hello ,kafka");
        //发送消息
        try {
            Object send = producer.send(record).get();
            log.info("【push message success】 send to topic-{} message-{}", TOPIC, JSONObject.toJSON(send));
        } catch (Exception e) {
            log.error("【push message failure】 meeage-{}", e);
        }
    }


}
